package com.coding4play.mqtt.store.impl;

import com.coding4play.mqtt.common.entity.session.ChannelSession;
import com.coding4play.mqtt.common.repository.ISessionRepository;
import com.coding4play.mqtt.store.config.RedisTemplateWrapper;
import com.coding4play.mqtt.store.util.SessionMap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelId;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.util.concurrent.GlobalEventExecutor;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 会话存储服务
 */
@Component
public class SessionRepositoryImpl extends RedisTemplateWrapper implements ISessionRepository {
    private final static String SET_CLIENT_SESSION_PRE = "mqtt:session:client_sessionInfo:";

    private ChannelGroup channelGroup;
    private Map<String, ChannelId> channelIdMap;

    public SessionRepositoryImpl(RedisTemplate<String, Object> redisTemplate) {
        super(redisTemplate);
        this.channelGroup = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
        this.channelIdMap = new ConcurrentHashMap<>();
    }

    @Override
    public void put(String clientId, ChannelSession channelSession, int expire) {
        //SessionStore对象不能正常转为JSON,使用工具类类解决
        Map sessionMap = SessionMap.transPublishToMapBeta(channelSession);
        if (expire > 0) {
            KV.set(getKey(clientId), sessionMap, Duration.ofSeconds(expire));
        } else {
            KV.set(getKey(clientId), sessionMap);
        }
    }

    @Override
    public void expire(String clientId, Integer expire) {
        redisTemplate.expire(getKey(clientId), Duration.ofSeconds(expire));
    }


    @Override
    public ChannelSession get(String clientId) {
        Object obj = KV.get(getKey(clientId));
        if (Objects.nonNull(obj)) {
            Map sessionMap = (HashMap) obj;
            return SessionMap.mapTransToPublishMsgBeta(sessionMap);
        }
        return null;
    }

    @Override
    public Boolean hasSession(String clientId) {
        return redisTemplate.hasKey(getKey(clientId));
    }

    @Override
    @Async
    public void delete(ChannelSession store) {
        ChannelId channelId = this.channelIdMap.remove(getMemoryMapKey(store.getBrokerId(), store.getChannelId()));
        if (channelId != null) {
            Channel channel = this.channelGroup.find(channelId);
            if (channel != null){
                this.channelGroup.remove(channel);
                channel.close();
            }
        }
        String clientId = store.getClientId();
        if (StringUtils.hasLength(clientId)) {
            this.redisTemplate.delete(getKey(clientId));
        }
    }

    @Override
    public ChannelGroup channelGroup() {
        return this.channelGroup;
    }

    @Override
    public void channelToMemoryMap(String brokerId, Channel channel) {
        this.channelIdMap.put(getMemoryMapKey(brokerId, channel.id().asLongText()), channel.id());
    }

    public Channel getChannel(ChannelSession store) {
        String memoryMapKey = getMemoryMapKey(store.getBrokerId(), store.getChannelId());
        return this.channelGroup.find(this.channelIdMap.get(memoryMapKey));
    }

    private String getMemoryMapKey(String brokerId, String channelId) {
        return brokerId + "_" + channelId;
    }

    private String getKey(String clientId) {
        return SET_CLIENT_SESSION_PRE + clientId;
    }

}
